package dm.ass2.user;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import dm.ass1.io.ProfileWritable;
import dm.ass1.io.RatingWritable;
import dm.ass2.ClusteringDriver;
import dm.ass2.io.CodebookItemsWritable;
import dm.ass2.io.CodebookUsersWritable;
import dm.ass2.user.io.UserMapOutputValueWritable;

public class UserDriver extends Configured implements Tool {
	private static final Log LOG = LogFactory.getLog(UserDriver.class);
	
	public static final String NUM_OF_USER_CLUSTERS = "num.of.user.clusters";
	public static final String NUM_OF_ITEM_CLUSTERS = "num.of.item.clusters";

	Set<Integer> itemsIds;
	Set<Integer> usersIds;
	
	//int numberOfUsers = -1;
	//int numberOfItems = -1;
	
	int numberOfUsersClusters = 2;
	int numberOfItemsClusters = 2;
	
	/**
	 * Generate the items code book, denoted V in the algorithm.
	 * 
	 * @param inputFileName The ratings file from MovieLens
	 * @throws IOException
	 */
	
	public void generateV(String inputFileName, String numOfItemsCluster, String numOfUsersClusters) throws IOException {
		this.numberOfItemsClusters = Integer.parseInt(numOfItemsCluster);
		this.numberOfUsersClusters = Integer.parseInt(numOfUsersClusters);
		
		Random randomGen = new Random(System.currentTimeMillis()); 
		countUsersAndItems(inputFileName);
		
		Map<Integer, CodebookItemsWritable> cbItems = new HashMap<Integer, CodebookItemsWritable>(itemsIds.size());
		CodebookItemsWritable cbItem;
		
		//for (int i = 0; i < numberOfItems; i++) {
		for(Integer i : itemsIds) {
			int itemClusterId = randomGen.nextInt(numberOfItemsClusters);
			cbItem = cbItems.get(itemClusterId);
			if (cbItem == null) {
				int[] itemsId = new int[0];
				float[] userClusterRatings = new float[numberOfUsersClusters];
				for (int j = 0; j < userClusterRatings.length; j++) {
					userClusterRatings[j] = randomGen.nextInt(4) + 1;
				}
				cbItem = new CodebookItemsWritable(itemClusterId, itemsId, userClusterRatings);
				cbItems.put(itemClusterId, cbItem);
			}
			
			cbItem.setItemIds(ArrayUtils.add(cbItem.getItemIds(), i));
		}
		
		writeVToFile(cbItems.values());
	}

	private void writeVToFile(Collection<CodebookItemsWritable> values) throws IOException {
		Path cbItemsPath = new Path(ClusteringDriver.OUTPUT_DIR + ClusteringDriver.CB_ITEMS_FILENAME);
		FileSystem fs;

		fs = FileSystem.get(new Configuration());

		if (fs.exists(cbItemsPath)) {
			fs.delete(cbItemsPath, true);
		}

		FSDataOutputStream out = fs.create(cbItemsPath);
		for (CodebookItemsWritable cbItem : values) {
			cbItem.write(out);
		}
		out.flush();
		out.close();

	}

	/**
	 * @param inputFileName
	 * @throws IOException
	 */
	private void countUsersAndItems(String inputFileName) throws IOException {
		itemsIds = new HashSet<Integer>();
		usersIds = new HashSet<Integer>();

		Path ratingsFile = new Path(inputFileName);
		FileSystem fs = FileSystem.get(new Configuration());

		BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(ratingsFile)));
		try {
			int lineCounter = 0;
			String line;
			line = br.readLine();
			while (line != null) {
				lineCounter++;
				String[] record = line.toString().split("\\W+");
				int user = Integer.parseInt(record[0]);
				int item = Integer.parseInt(record[1]);
				itemsIds.add(item);
				usersIds.add(user);

				line = br.readLine();
			}
		} finally {
			br.close();
		}
		
		//numberOfItems = itemsIds.size();
		//numberOfUsers = usersIds.size();

		LOG.info(String.format("Input Data: number of users: %d, number of items %d.", usersIds.size(), itemsIds.size()));
	}
	
	private boolean runProfilingJob(String inputPath) throws Exception {
		Job job = new Job(new Configuration());
		Configuration conf = job.getConfiguration();
		conf.set(NUM_OF_ITEM_CLUSTERS, String.valueOf(this.numberOfItemsClusters));
		conf.set(NUM_OF_USER_CLUSTERS, String.valueOf(this.numberOfUsersClusters));
		
		FileSystem fs = FileSystem.getLocal(conf);
	    Path tempOutputPath = new Path("output/userTemp/");
	    if (fs.exists(tempOutputPath)) {
	    	fs.delete(tempOutputPath,true);			
		}
	    		
		job.setJarByClass(UserDriver.class);
		FileInputFormat.addInputPath(job, new Path(inputPath));
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		SequenceFileOutputFormat.setOutputPath(job, tempOutputPath);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(CodebookUsersWritable.class);
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(UserMapOutputValueWritable.class);
		job.setMapperClass(UserMap.class);
		job.setCombinerClass(UserCombiner.class);
		job.setReducerClass(UserReducer.class);
		
		return job.waitForCompletion(true);		
	}
	
	@Override
	public int run(String[] args) throws Exception {
		//generateV(args[0]);
		runProfilingJob(args[0]);
		
		return 0;

	}

	public static void main(String[] args) throws Exception {
		int ret = ToolRunner.run(new UserDriver(), args);
		System.exit(ret);
	}
}
